package com.han.elasticsearch.kafka;

import com.han.elasticsearch.entity.ElasticSearchUser;
import com.han.elasticsearch.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * @author hanjl
 * @Description kafka消费者
 * @ClassName KafkaConfumer
 * @createTime 2021/12/22 17:01
 */
@Component
@Slf4j
public class KafkaConsumer {

    /**
     * 当发送过来的数据是个集合时，需要list批量接受消息
     * @param records
     * @param acknowledgment
     */
    @KafkaListener(topics = {"kafka-user"})
    public void consumerMessage(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment){

        for (ConsumerRecord<String,String> record : records){
            Optional<String> value = Optional.ofNullable(record.value());
            if(value.isPresent()){
                String messages = value.get();
                List<ElasticSearchUser> users = JsonUtil.toList(messages, ElasticSearchUser.class);
                System.out.println("消费者消费了：" + users.toString());
            }
        }
    }

    /**
     * 接受单条消息
     * @param record
     * @param acknowledgment
     */
    @KafkaListener(topics = {"kafka-user2"})
    public void consumerMessage2(ConsumerRecord<String, String> record, Acknowledgment acknowledgment){
        Optional<String> value = Optional.ofNullable(record.value());
        if(value.isPresent()){
            String messages = value.get();
            List<ElasticSearchUser> users = JsonUtil.toList(messages, ElasticSearchUser.class);
            System.out.println("消费者消费了：" + users.toString());
        }
    }
}
